package com.lolineet.mqtt.conf;

import com.lolineet.mqtt.properties.MqttProperties;
import com.lolineet.mqtt.service.MqttMessageReceiver;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;

/**
 * @Auther lolineet
 * @Email tj.zhanghongwei@163.com
 * @Date 2021/11/23 14:47
 * @Description
 */
@Configuration
@IntegrationComponentScan
@ConditionalOnProperty(
        prefix = "lolineet.mqtt",
        name = "enable",
        havingValue = "true"
)
public class MqttInboundConfiguration {

    private MqttProperties mqttProperties;
    private MqttPahoClientFactory factory;

    public MqttInboundConfiguration(MqttProperties mqttProperties, MqttPahoClientFactory factory) {
        this.mqttProperties = mqttProperties;
        this.factory = factory;
    }

    /**
     * 此处可以使用其他消息通道
     * Spring Integration默认的消息通道，它允许将消息发送给一个订阅者，然后阻碍发送直到消息被接收。
     *
     * @return
     */
    @Bean
    public MessageChannel mqttInBoundChannel() {
        return new DirectChannel();
    }

    /**
     * 适配器, 两个topic共用一个adapter
     * 客户端作为消费者，订阅主题，消费消息
     *
     * @param
     * @param
     * @return
     */
    @Bean//name = MqttConfiguration.ADAPTER_NAME_IN)
    public MessageProducerSupport mqttInbound() {

        MqttPahoMessageDrivenChannelAdapter
                adapter = new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getClientId() + System.currentTimeMillis(),
                factory,
                mqttProperties.getTopics());
        adapter.setCompletionTimeout(60000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setRecoveryInterval(10000);
        adapter.setQos(0);
        adapter.setOutputChannel(mqttInBoundChannel());
        return adapter;
    }


    //    /**
//     * mqtt入站消息处理工具，对于指定消息入站通道接收到生产者生产的消息后处理消息的工具。
//     * @return
//     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInBoundChannel")
    public MqttMessageReceiver mqttMessageReceiver() {
        return new MqttMessageReceiver();
    }
}